/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.trp.report.sort;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class ParellelMergeSort<T> {

	private static class SubMerger<T> implements DataProvider<T> {
		private MergeSort<T> sorter;
		private Thread thread;
		private Queue<T> queue;
		private boolean done;
		private ReentrantLock lock;
		private Condition put;
		private Condition get;
		private int maxSize;

		public SubMerger(List<DataProvider<T>> providers,
		        Comparator<T> comparator, int maxSize) {
			this.sorter = new MergeSort<T>(providers, comparator);
			this.maxSize = maxSize;
			this.queue = new LinkedList<>();
			this.lock = new ReentrantLock();
			this.put = lock.newCondition();
			this.get = lock.newCondition();
			this.done = false;

			init();

		}

		private void init() {
			thread = new Thread(new Runnable() {

				@Override
				public void run() {

					T obj;
					while ((obj = sorter.poll()) != null) {

						lock.lock();
						try {

							while (queue.size() >= maxSize) {
								try {
									put.await();
								} catch (InterruptedException e) {
									done = true;
									get.signal();
									return;
								}
							}
							queue.add(obj);
							get.signal();
						} finally {
							lock.unlock();
						}
					}

					lock.lock();
					try {
						done = true;
						get.signal();
					} finally {
						lock.unlock();
					}
				}

			});

			thread.start();
		}

		@Override
		public T peek() {

			lock.lock();
			try {

				while (queue.isEmpty()) {
					if (done)
						return null;
					try {
						get.await();
					} catch (InterruptedException e) {
						return null;
					}
				}
				return queue.peek();
			} finally {
				lock.unlock();
			}

		}

		@Override
		public T poll() {
			lock.lock();
			try {
				while (queue.isEmpty()) {
					if (done)
						return null;
					try {
						get.await();
					} catch (InterruptedException e) {
						return null;
					}
				}
				T obj = queue.poll();
				if (queue.size() < maxSize)
					put.signal();
				return obj;
			} finally {
				lock.unlock();
			}
		}
	}

	private MergeSort<T> merger;

	public ParellelMergeSort(List<DataProvider<T>> providers,
	        Comparator<T> comparator, int threadNum, int maxSize) {

		if (providers == null || providers.isEmpty()) {
			throw new RuntimeException("no data provider");
		}

		if (comparator == null) {
			throw new RuntimeException("no comparator");
		}

		if (threadNum <= 1 || threadNum > providers.size()) {
			throw new RuntimeException("Incorrect thread num");
		}

		List<List<DataProvider<T>>> lists = new ArrayList<>();

		for (int i = 0; i < threadNum; i++) {
			lists.add(new ArrayList<DataProvider<T>>());
		}

		for (int i = 0; i < providers.size(); i++) {
			lists.get(i % threadNum).add(providers.get(i));
		}

		List<DataProvider<T>> subMergers = new ArrayList<>();

		for (int i = 0; i < threadNum; i++) {
			subMergers.add(new SubMerger<T>(lists.get(i), comparator, maxSize));
		}

		this.merger = new MergeSort<T>(subMergers, comparator);
	}

	public T peek() {
		return merger.peek();
	}

	public T poll() {
		return merger.poll();
	}

	private static class TestProvider implements DataProvider<Integer> {

		private int[] data = null;
		private int index = 0;
		private int size = 0;

		public TestProvider(int size) {
			this.data = new int[size];
			this.size = size;
			Random rand = new Random();
			for (int i = 0; i < size; i++) {
				data[i] = rand.nextInt();
			}

			Arrays.sort(data);
		}

		@Override
		public Integer peek() {
			if (index >= size)
				return null;
			return data[index];
		}

		@Override
		public Integer poll() {
			if (index >= size)
				return null;
			int ret = data[index];
			index++;
			return ret;
		}

	}

	public static void main(String[] args) {

		List<DataProvider<Integer>> providers = new ArrayList<>();

		for (int i = 0; i < 100; i++) {
			providers.add(new TestProvider(10000));
		}

		long start = System.currentTimeMillis();
		ParellelMergeSort<Integer> sorter = new ParellelMergeSort<Integer>(
		        providers, new Comparator<Integer>() {

			        @Override
			        public int compare(Integer o1, Integer o2) {
				        return o1.compareTo(o2);
			        }

		        }, 2, 10000);

		List<Integer> results = new ArrayList<>();
		Integer obj;
		while ((obj = sorter.poll()) != null) {
			results.add(obj);
		}

		System.out.println((System.currentTimeMillis() - start) + " ms");
		System.out.println(results.size());

		boolean flag = true;

		for (int i = 1; i < results.size(); i++) {
			if (results.get(i - 1) > results.get(i)) {
				flag = false;
				break;
			}
		}

		System.out.println(flag ? "OK" : "ERROR");

	}

}
